package com.shujia.streaming

import org.apache.spark.SparkContext
import org.apache.spark.sql.SparkSession
import org.apache.spark.streaming.dstream.ReceiverInputDStream
import org.apache.spark.streaming.{Durations, StreamingContext}

object Demo6Submit {
  def main(args: Array[String]): Unit = {
    val sparkSession: SparkSession = SparkSession.builder()
      .appName("提交命令执行")
      .config("spark.sql.shuffle.partitions", 1)
      .getOrCreate()

    val sparkContext: SparkContext = sparkSession.sparkContext
    val streamingContext = new StreamingContext(sparkContext, Durations.seconds(5))

    val linesDS: ReceiverInputDStream[String] = streamingContext.socketTextStream("master", 12345)

    linesDS
      .flatMap(_.split(" "))
      .map((_,1))
      .reduceByKey(_+_)
      .print()


    streamingContext.start()
    streamingContext.awaitTermination()
    streamingContext.stop()
  }
}
